package com.atguigu.aggreagte;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gmd
 * @desc keyBy（按键分区）功能演示
 * @since 2024-11-25 16:36:11
 */
public class KeyByDemo {

    public static void main2(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s2", 22L, 22),
                new WaterSensor("s3", 33L, 33),
                new WaterSensor("s3", 22L, 22)
        );

        /*
         * 按照id分组的keyBy功能演示
         *
         * @note keyBy的作用是将数据按照指定的key进行分组，返回的是一个KeyedStream，键控流
         *
         * @important
         *  1. keyBy不是一个转换算子，它只是对数据进行重分区，不能设置并行度
         *  2. 分组与分区的关系：
         *     1）keyBy是对数据进行分组，保证相同key的数据在同一个分区（子任务）
         *     2）分区：一个子任务可以理解为一个分区，一个分区（子任务）中可以存在多个分组（key）
         */
        KeyedStream<WaterSensor, String> sensorKS = sensorDS
                .keyBy(new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor value) throws Exception {
                        // 自定义分组规则
                        return value.getId();
                    }
                });

        /*
         * 观察程序运行结果，发现并没有体现分区？
         * keyBy 操作在Flink中只对逻辑上进行了分组，而并没有直接影响数据的物理输出顺序。KeyedStream 的分组效果体现在它的后续处理操作中，
         * 比如 window、aggregate、reduce 等，这些操作会基于 Key 进行数据处理。
         * 在你的代码中，虽然你使用了 keyBy 对 WaterSensor 进行了逻辑上的分组，但由于没有进一步的聚合、窗口或者其他算子操作，
         * 数据仍然是按源数据流的顺序进行输出。print() 操作只是按物理并行度输出数据，并没有对逻辑分组结果进行处理，所以看不到明显的分组效果。
         */
        sensorKS.print();
        env.execute();
    }


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s2", 22L, 22),
                new WaterSensor("s3", 33L, 33),
                new WaterSensor("s3", 22L, 22)
        );

        KeyedStream<WaterSensor, String> sensorKS = sensorDS
                .keyBy(new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor value) throws Exception {
                        // 自定义分组规则
                        return value.getId();
                    }
                });

        // 使用 reduce 算子来验证 keyBy 的效果
        sensorKS.reduce((value1, value2) ->
                new WaterSensor(value1.getId() + "-" + value2.getId(), 0L, 0)
        ).print();

        env.execute();
    }

}
